package org.jetlinks.community.messaging.kafka.impl;

import org.jetlinks.community.messaging.kafka.KafkaConsumer;
import org.jetlinks.community.messaging.kafka.KafkaDataSource;
import org.jetlinks.community.messaging.kafka.ReactorKafkaConsumer;
import org.jetlinks.community.messaging.kafka.command.CreateConsumer;
import org.jetlinks.community.datasource.CommandHandler;
import org.jetlinks.community.datasource.DataSource;
import reactor.core.publisher.Mono;

class CreateConsumerHandler implements CommandHandler<CreateConsumer, Mono<KafkaConsumer>> {

     static CreateConsumerHandler INSTANCE = new CreateConsumerHandler();

    @Override
    public Mono<KafkaConsumer> handle(DataSource dataSource, CreateConsumer command) {
        KafkaDataSource kafkaDataSource = dataSource.unwrap(KafkaDataSource.class);

        ReactorKafkaConsumer consumer = new ReactorKafkaConsumer(command.getTopics(),
                                                          command.getProperties() == null
                                                              ? kafkaDataSource.getConfig()
                                                              : command.getProperties());
        consumer.init();

        return Mono.just(consumer);
    }
}
